package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAddApp extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeCartAddApp().init(2003,"DwdTradeCartAddApp");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        readOdsDb(tEnv,"DwdTradeCartAddApp");

        //过滤出加购表数据cd
        Table carTable = tEnv.sqlQuery("select" +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['course_name'] course_name, " +
                " `data`['cart_price'] cart_price, " +
                " `data`['session_id'] session_wid, " +
                " `data`['create_time'] create_time, " +
                " `data`['update_time'] update_time, " +
                " `data`['deleted'] deleted, " +
                " `data`['sold'] sold, " +
                " ts, " +
                " xid " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='cart_info'"
        );
        //待后续优化
        //tEnv.createTemporaryView("cartTable",carTable);

        //把结果写入kafka
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                " id string, " +
                " user_id string, " +
                " course_name string, " +
                " cart_price string, " +
                " session_id string, " +
                " create_time string, " +
                " update_time string, " +
                " deleted string, " +
                " sold string, " +
                " ts String, " +
                " xid String " +
                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_TRADE_CART_ADD)
                );
        carTable.executeInsert("dwd_trade_cart_add");


    }
}
